package mapReduce.demo02_phoneTrafficCount;

import mapReduce.util.JobSubmit;
import mapReduce.writable.PhoneTrafficWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;

public class PhoneTrafficCountApp {

    public static class PTCMapper extends Mapper<LongWritable,Text,Text, PhoneTrafficWritable>{

        Text outKey = new Text();
        PhoneTrafficWritable outValue = new PhoneTrafficWritable();

        @Override
        protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, PhoneTrafficWritable>.Context context) throws IOException, InterruptedException {
            //把每行内容由Text类型转成String类型，方便后面使用
            String line = value.toString();
            //按照制表符进行分割，分割出每个字段组成的数组
            String[] fields = line.split("\t");
            //获取每行数据中的手机号
            String phoneNum = fields[1];
            //获取每行数据中的上行流量，下行流量 并计算出总流量
            int upTraffic = Integer.parseInt(fields[fields.length-3]);
            int downTraffic = Integer.parseInt(fields[fields.length-2]);
            int sumTraffic = upTraffic + downTraffic;

            //分别用相应的数据设置 输出的key和value
            outKey.set(phoneNum);
            outValue.setUpTraffic(upTraffic);
            outValue.setDownTraffic(downTraffic);
            outValue.setSumTraffic(sumTraffic);

            context.write(outKey,outValue);
        }
    }

    public static class PTCReducer extends Reducer<Text,PhoneTrafficWritable,Text,PhoneTrafficWritable>{

        PhoneTrafficWritable outValue = new PhoneTrafficWritable();

        @Override
        protected void reduce(Text key, Iterable<PhoneTrafficWritable> values, Reducer<Text, PhoneTrafficWritable, Text, PhoneTrafficWritable>.Context context) throws IOException, InterruptedException {
            /**
             * 数据格式
             *      138 {PhoneTrafficWritable,PhoneTrafficWritable,PhoneTrafficWritable,......}
             *      139 {PhoneTrafficWritable,PhoneTrafficWritable,PhoneTrafficWritable,......}
             *      ......
             */
            //初始化 上行流量，下行流量和总流量
            int upTrafficSum = 0;
            int downTrafficSum = 0;
            int sumTrafficSum = 0;

            //遍历每个手机号的手机流量，分别对上行 下行和总流量进行求和
            for (PhoneTrafficWritable value : values) {
                upTrafficSum += value.getUpTraffic();
                downTrafficSum += value.getDownTraffic();
                sumTrafficSum += value.getSumTraffic();
            }

            //封装求和之后的流量
            outValue.setUpTraffic(upTrafficSum);
            outValue.setDownTraffic(downTrafficSum);
            outValue.setSumTraffic(sumTrafficSum);

            context.write(key,outValue);

        }
    }

    public static void main(String[] args) {
        JobSubmit.submitBaseJob(PhoneTrafficCountApp.class,args);
    }
}
